using Pegasus.Diagnostics;
using System;
using System.Threading;

using Pegasus.Log4Net;

namespace Pegasus.Threading
{
	/// <summary>
	/// The MultiThreadedProcessQueue class extends the functionality of the ProcessQueue class.
	/// Like the ProcessQueue class, this class is a FIFO queue for which the user may register a
	/// callback to process requests or objects in the background as they are dequeued.
	/// Unlike the ProcessQueue, instead of being limited to just one dedicated thread to process
	/// the requests or objects, the MultiThreadedProcessQueue employs an entire thread pool.
	/// 
	/// Once again, the user can queue up to as many objects as needed and they will be 
	/// processed on a background thread.  However, because the thread pool is
	/// initialized to a maximum number of worker threads (to avoid performance problems), if
	/// there are more objects or requests than there are available worker threads, reading 
	/// from the queue will block until a worker thread becomes available.
	/// </summary>
	public class MultiThreadedProcessQueue : ProcessQueue
	{
		// Local Instance Values
		private ILog m_log = LogManager.GetLogger( typeof( MultiThreadedProcessQueue ) );
		private AppThreadPool m_appThreadPool = null;

		/// <summary>
		/// Get/Set the number of worker threads.
		/// </summary>
		/// <remarks>
		/// This can be used to throttle the queue.
		/// </remarks>
		public int WorkerThreads
		{
			get
			{
				return m_appThreadPool.MaximumThreads;
			}

			set
			{
				m_appThreadPool.MaximumThreads = value;
			}
		}
		
		/// <summary>
		/// Basic constructor expects an integer that sets the maximum number of worker threads
		/// in the AppThreadPool used to process items from the queue.
		/// </summary>
		/// <param name="maxWorkerThreads">An integer greater than or equal to 1.</param>
		public MultiThreadedProcessQueue( int maxWorkerThreads ) : base()
		{
			if( maxWorkerThreads < 1 )
			{
				throw new ArgumentOutOfRangeException( "maxWorkerThreads", maxWorkerThreads, "The number of worker threads you specify must be greater than or equal to 1." );
			}

			m_appThreadPool = new AppThreadPool();
			m_appThreadPool.MaximumThreads = maxWorkerThreads;
		}

		/// <summary>
		/// Stop processing objects in the queue.  The "wait" argument indicates
		/// whether or not we should wait for the queue to be empty before 
		/// we stop processing.  Even if we choose not to wait for the queue
		/// to empty, this will block until all the currently working threads
		/// in the thread pool have finished processing their objects.
		/// </summary>
		/// <param name="wait"></param>
		public override void Stop( bool wait )
		{
			if( wait )
			{
				while( m_objectQueue.Count > 0 )
				{
					// just loop and yeld thread
					Thread.Sleep( 1 );
				}
			}

			m_running = false;
			m_newObjEvent.Set();
			m_appThreadPool.AbortAllThreads();
		}

		/// <summary>
		/// The background thread for processing the objects in the queue.
		/// </summary>
		protected override void ThreadProc()
		{
			while( m_running )
			{
				object obj = null;

				// Lock the queue while we get the next object
				lock( m_objectQueue.SyncRoot )
				{
					// Get the object from the queue.  If there are no object in the queue
					// and we call the Dequeue() method it will throw.
					if( m_objectQueue.Count > 0 )
					{
						obj = m_objectQueue.Dequeue();
						m_newObjEvent.Reset();
					}
				}

				if( obj != null )
				{
					try
					{
						m_appThreadPool.QueueUserWorkItem( m_processHandler, obj );
					}
					catch( Exception e )
					{
						m_log.Error( "Unhandled exception while processing object in ProcessQueue class", e );
					}
				}
				else
				{
					m_newObjEvent.WaitOne();
				}
			}
		}
	}
}
